package parallels.servertech.server.nio.event;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.logging.Logger;

import parallels.servertech.server.nio.ChannelSelector;
import parallels.servertech.server.nio.event.ChannelEvent.ChannelEventType;

// multi-threaded dispatcher with inbound events queue ( reading ready ) and outboud events queue ( write ready )
// selection thread will put filled buffers wrapped as events into inboud queue, ChannelEventDispather will dispatch events to listeners
// listeners can fire events by setting outboud buffers on them. dispatcher will place such events to outboud queue
// selection thread will pull events from outbound queue and write results back to clients
// outboud doesn't mean ready for reading ??? just processed events queue
// 1. use single thread executor for dispatcer to simplify evolution

// put all former decoder logic into AsyncServer
// name selection thread as selector
public class ChannelEventDispatcher {
	private static final Logger log = Logger.getLogger("event-dispatcher");
	
	private static final int DEFAULT_EVENT_QUEUE_CAPACITY = 1024;
	
	private final ConcurrentMap<Integer, ChannelEventHandler> handlers;
	private final Map<Integer, ChannelEvent> replyEvents = new HashMap<>();
	private final BlockingQueue<ChannelEvent> eventQueue = new LinkedBlockingQueue<>(DEFAULT_EVENT_QUEUE_CAPACITY);
	
	private final ChannelSelector channelSelector;
	private final Thread eventProcessor;
	
	private volatile boolean started = false;
	
	private class EventProcessor implements Runnable {
		@Override
		public void run() {
			try {
				while (started) {
					// TODO use poll with timeout
					ChannelEvent event = eventQueue.take();
					SocketChannel channel = event.getChannel();
					log.info(event.getType() + " event received for channel @" + channel.hashCode());
					
					ChannelEventHandler handler = handlers.get(channel.hashCode());
					if (handler == null) {
						log.info("event processing skipped for channel @" + channel.hashCode());
						continue;
					}
					
					try {
						handleEvent(event, handler);
					} catch (IOException ioe) {
						log.severe("event handling failure: " + ioe.getClass() + "(" + ioe.getMessage() + ")");
					}
				}
			} catch (InterruptedException ie) {
				log.info("event processor interrupted");
				return;
			}
		}

		private void handleEvent(ChannelEvent event, ChannelEventHandler handler) throws IOException {
			ChannelEvent handlingEvent = event;
			while (started && handlingEvent != null) {
				ChannelEventType eventType = handlingEvent.getType();
				SocketChannel channel = handlingEvent.getChannel();
				
				log.info("handling channel @" + channel.hashCode() + " event: " + handlingEvent.getType());
				switch (eventType) {
				case CHANNEL_READY_FOR_READING:
					handler.channelReadable(handlingEvent);
					break;
				case INPUT_BUFFER_READY_FOR_FILLING:
					ByteBuffer inputBuffer = handler.getChannelContext().getBuffer();
					int bytesRead = channel.read(inputBuffer);
					
					log.info(bytesRead + " bytes read from channel @" + channel.hashCode());
					if (bytesRead > 0) {
						inputBuffer.flip();
						handler.inputBufferFilled(new ChannelEvent(handlingEvent.getEventDispatcher(), ChannelEventType.INPUT_BUFFER_FILLED, channel, handlingEvent.getSelectionKey()));
					} else {
						log.info("no more bytes available");
						SelectionKey selectKey = handlingEvent.getSelectionKey();
						selectKey.interestOps(selectKey.interestOps() | SelectionKey.OP_READ);
					}
					
					break;
				case OUTPUT_BUFFER_READY_FOR_DRAINING:
					log.info(handler.getChannelContext().getBuffer().remaining() + " bytes available for writing");
					handlingEvent.getSelectionKey().interestOps(handlingEvent.getSelectionKey().interestOps() | SelectionKey.OP_WRITE);
					handlingEvent.getSelectionKey().selector().wakeup();
					// TODO store buffers in ChannelContext for processing later
					break;
				case CHANNEL_READY_FOR_WRITING:
					int bytesWritten = channel.write(handler.getChannelContext().getBuffer());
					log.info(bytesWritten + " bytes written to channel");
					if (!handler.getChannelContext().getBuffer().hasRemaining()) {
						// TODO wrap in channel context higher level methods addInterestOp()/removeInterestOp()
						handlingEvent.getSelectionKey().interestOps(handlingEvent.getSelectionKey().interestOps() | SelectionKey.OP_READ);
						handler.outputBufferDrained(new ChannelEvent(handlingEvent.getEventDispatcher(), ChannelEventType.OUTPUT_BUFFER_DRAINED, channel, handlingEvent.getSelectionKey()));
					}
					break;
				default:
					log.info("unsupported event type: " + eventType);
				}
				
				handlingEvent = replyEvents.remove(handlingEvent.getChannel().hashCode());
			}
		}
	}
	
	public ChannelEventDispatcher(ChannelSelector channelSelector) {
		handlers = new ConcurrentHashMap<>();
		eventProcessor = new Thread(new EventProcessor(), "event-processor");
		this.channelSelector = channelSelector;
	}

	public void start() {
		started = true;
		eventProcessor.start();
		log.info("dispatcher started");
	}
	
	public void stop() {
		log.info("stopping dispatcher");
		started = false;
		eventProcessor.interrupt();
		channelSelector.stop();
	}

	public void addEventHandler(SelectionKey selectKey, ChannelEventHandler listener) {
		SocketChannel channel = (SocketChannel) selectKey.channel();
		log.info("channel @" + channel.hashCode() + " event listener registered");
		handlers.put(channel.hashCode(), listener);
	}

	public boolean sendEvent(ChannelEvent event) {
		SocketChannel channel = event.getChannel();
		boolean enqueued = eventQueue.offer(event);
		log.info("channel @" + channel.hashCode() + " event " + event.getType() + " enqueued: " + enqueued);
		if (enqueued) {
			switch (event.getType()) {
			case CHANNEL_READY_FOR_READING:
				event.getSelectionKey().interestOps(event.getSelectionKey().interestOps() & ~SelectionKey.OP_READ);
				break;
			case CHANNEL_READY_FOR_WRITING:
				event.getSelectionKey().interestOps(event.getSelectionKey().interestOps() & ~SelectionKey.OP_WRITE);
				break;
			default:
				// left interest ops intact	
			}
			
		}
		return enqueued;
	}

	public void sendReply(ChannelEvent event) {
		// TODO drop reply handling logic
		// add ChannelContext with methods setInputBufferReadyForFilling()
		// setChannelReadyForWriting()
		// setOutputBufferReadyForDraining()
		// use dedicated ChannelEventHandlingState instead of putting all states into event types
		SocketChannel channel = event.getChannel();
		replyEvents.put(channel.hashCode(), event);
		log.info("channel @" + channel.hashCode() + " reply event stored");
	}
}
